package com.hlz.flink.chapter05;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 简单聚合操作2
 *
 * @author Hongliang Zhu
 * @create 2023-05-04 23:33
 */
public class SimpleAggregation {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> streamSource = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Alice", "./prod?id=100", 3000L),
                new Event("Bob", "./home", 4000L),
                new Event("Bob", "./prod?id=9", 5000L),
                new Event("Bob", "./prod?id=2", 6000L),
                new Event("Mary", "./prod?id=3", 7000L)
                );

        // 按键分组后聚合,提取当
        streamSource.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event value) throws Exception {
                return value.getUser();
            }
        }).max("timeStamp")
                .print("max: ");

        streamSource.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event value) throws Exception {
                return value.getUser();
            }
        }).maxBy("timeStamp")
                .print("maxBy: ");


        env.execute();

    }
}
